# coding:utf8
# 导包
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    monotonically_increasing_id, col, when, lit, expr, isnull, coalesce, udf
)
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, FloatType
import re


def remove_trailing_slashes(text):
    """
    此函数使用正则表达式去除字符串头尾的 " / " 组合
    """
    if isinstance(text, str):
        # 使用正则表达式去除开头和结尾的 " / " 组合
        cleaned_text = re.sub(r"^(\s*/\s*)+|(\s*/\s*)+$", "", text)
        return cleaned_text
    else:
        return text


if __name__ == '__main__':
    # 创建SparkSession对象，设置应用名称为'sparkSQL'，以本地模式运行，使用所有可用核心
    # 配置shuffle的分区数为2，指定Spark SQL的仓库目录为Hive仓库目录，设置Hive元数据服务的URI
    # 并启用Hive支持
    spark = SparkSession.builder.appName('sparkSQL').master('local[*]'). \
        config("spark.sql.shuffle.partitions", 2). \
        config("spark.sql.warehouse.dir", "hdfs://node1.itcast.cn:8020/user/hive/warehouse"). \
        config("hive.metastore.uris", "thrift://node1.itcast.cn:9083"). \
        enableHiveSupport().getOrCreate()

    # 构建数据结构模式，用于读取CSV文件
    schema = StructType().add("type", StringType(), nullable=True). \
        add("title", StringType(), nullable=True). \
        add("companyTitle", StringType(), nullable=True). \
        add("minSalary", IntegerType(), nullable=True). \
        add("maxSalary", IntegerType(), nullable=True). \
        add("workExperience", StringType(), nullable=True). \
        add("educational", StringType(), nullable=True). \
        add("detailUrl", StringType(), nullable=True). \
        add("companyPeople", StringType(), nullable=True). \
        add("workTag", StringType(), nullable=True). \
        add("welfare", StringType(), nullable=True). \
        add("imgSrc", StringType(), nullable=True). \
        add("city", StringType(), nullable=True)

    # 从HDFS读取CSV文件，指定分隔符为逗号，引用字符和转义字符为双引号，有表头，编码为UTF-8，并使用自定义的模式
    # 这里将读取本地文件的路径修改为HDFS路径，例如假设文件在HDFS的/spark_data/jobData.csv位置，可根据实际情况调整路径
    df = spark.read.format("csv"). \
        option("sep", ","). \
        option("quote", '"'). \
        option("escape", '"'). \
        option("header", True). \
        option("encoding", "utf-8"). \
        schema(schema=schema). \
        load("hdfs://node1.itcast.cn:8020/SparkAnalyzeRecruitmentSolution/jobData.csv")

    # 删除重复的数据行
    new_df = df.drop_duplicates(subset=['title', 'companyTitle', 'maxSalary', 'minSalary'])

    # 显示数据框的内容
    new_df.show()
    # 打印new_df行数
    print("new_df的行数为：", new_df.count())

    # 为数据框添加一个自增列名为'id'
    new_df = new_df.withColumn('id', monotonically_increasing_id().cast(IntegerType()))

    # 将workExperience字段中值为应届生的全部替换为“在校/应届”
    new_df = new_df.withColumn("workExperience",
                               when(col("workExperience") == "应届生", "在校/应届")
                            .otherwise(col("workExperience")))

    # 创建工作经验到数值的映射字典
    experience_mapping = {
        "经验不限": 1,
        "在校/应届": 2,
        "1个月": 3,
        "2个月": 4,
        "3个月": 5,
        "4个月": 6,
        "5个月": 7,
        "6个月": 8,
        "7个月": 9,
        "8个月": 10,
        "9个月": 11,
        "10个月": 12,
        "11个月": 13,
        "12个月": 14,
        "1年以内": 15,
        "1-3年": 16,
        "3-5年": 17,
        "5-10年": 18,
        "10年以上": 19
    }

    # 使用when和otherwise方法添加experience_value列
    new_df = new_df.withColumn("experience_value",
                               when(col("workExperience").isin(list(experience_mapping.keys())), col("workExperience")))
    for key, value in experience_mapping.items():
        new_df = new_df.withColumn("experience_value",
                                   when(col("workExperience") == key, value).otherwise(col("experience_value")))

    # 确保experience_value列为整型数据
    new_df = new_df.withColumn("experience_value", col("experience_value").cast(IntegerType()))

    # 处理minSalary列，将小于等于1000的数据乘以26替换原来数据
    new_df = new_df.withColumn("minSalary",
                               when(col("minSalary") <= 1000, col("minSalary") * 26).otherwise(col("minSalary")))
    # 处理maxSalary列，将小于等于1000的数据乘以26替换原来数据
    new_df = new_df.withColumn("maxSalary",
                               when(col("maxSalary") <= 1000, col("maxSalary") * 26).otherwise(col("maxSalary")))

    # 处理workExperience字段的缺失值，使用coalesce函数，用'经验不限'填充（可根据实际需求改）
    new_df = new_df.withColumn("workExperience", coalesce(col("workExperience"), lit("经验不限")))

    # 处理educational字段的缺失值，使用coalesce函数，用'无学历要求'填充（可根据实际需求改）
    new_df = new_df.withColumn("educational", coalesce(col("educational"), lit("学历不限")))

    # 处理companyPeople字段的缺失值，使用coalesce函数，用'未知规模'填充（可根据实际需求改）
    new_df = new_df.withColumn("companyPeople", coalesce(col("companyPeople"), lit("0-20")))

    # 处理workTag字段的缺失值，使用coalesce函数，用'无岗位标签'填充（可根据实际需求改）
    new_df = new_df.withColumn("workTag", coalesce(col("workTag"), lit("")))

    # 处理welfare字段的缺失值，使用coalesce函数，用'无福利'填充（可根据实际需求改）
    new_df = new_df.withColumn("welfare", coalesce(col("welfare"), lit("福利少")))


    # 注册用户自定义函数 (UDF)
    from pyspark.sql.types import StringType
    remove_trailing_slashes_udf = udf(remove_trailing_slashes, StringType())


    # 对 workTag 列应用函数进行处理
    new_df = new_df.withColumn("workTag", remove_trailing_slashes_udf(new_df["workTag"]))


    # 将数据框写入 MySQL 数据库，指定连接 URL、表名、用户名、密码和编码
    new_df.write.mode("overwrite").format("jdbc"). \
        option("url", "jdbc:mysql://node1.itcast.cn:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "jobData"). \
        option("user", "root"). \
        option("password", "123456"). \
        option("encoding", "utf-8").save()

    # 将数据框以覆盖模式保存为Hive表，表名为'jobData'，存储格式为Parquet
    new_df.write.mode("overwrite").saveAsTable("jobData", "parquet")

    # 使用Spark SQL查询Hive表'jobData'并显示结果
    spark.sql("select workTag from jobData").show()